#include "config.h"

#include <sys/types.h>
#include <regex.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>
#include <string.h>
#include <errno.h>
#include <pthread.h>
#include <signal.h>
#include <sys/vfs.h>
#include <unistd.h>
#include <getopt.h>
#include <sys/statvfs.h>

#define DBG_SUBSYS S_LIBCLUSTER

#include "sysy_lib.h"
#include "env.h"
#include "core.h"
#include "fence.h"
#include "configure.h"
#include "lichconf.h"
#include "job_dock.h"
#include "minirpc.h"
#include "rpc_proto.h"
#include "metadata.h"
#include "ynet_rpc.h"
#include "net_global.h"
#include "net_vip.h"
#include "etcd.h"
#include "node.h"
#include "ylog.h"
#include "dbg.h"
#include "nodectl.h"

#define ROOTABLE "rootable"

extern int cluster_rpc_init();

node_t __node__;
static int __node_deleting = 0;
static sy_spinlock_t __stat_lock__;

int node_create_workdir(const char *home, const char *clustername)
{
        int ret;
        char path[MAX_PATH_LEN];

        ret = path_validate(home, YLIB_ISDIR, YLIB_DIRCREATE);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        snprintf(path, MAX_NAME_LEN, "%s/config/clustername", home);

        ret = _set_value(path, clustername, strlen(clustername), O_CREAT | O_EXCL | O_SYNC);
        if (unlikely(ret)) {
                if (ret == EEXIST) {
                        DWARN("%s already inited\n", home);
                }

                GOTO(err_ret, ret);
        }

        snprintf(path, MAX_NAME_LEN, "%s/config/name", home);

        ret = _set_value(path, "none", strlen("none"), O_CREAT | O_EXCL | O_SYNC);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        snprintf(path, MAX_NAME_LEN, "%s/config/status", home);

        ret = _set_value(path, NODE_STATUS_NONE, strlen(NODE_STATUS_NONE),
                         O_CREAT | O_EXCL | O_SYNC);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        return 0;
err_ret:
        return ret;
}

static int __node_checkname(const char *name)
{
        int ret;
        char addr[MAX_NAME_LEN], host[MAX_NAME_LEN];

        ret = net_gethostname(host, MAX_BUF_LEN);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        if (strcmp(name, host)) {
                ret = EINVAL;
                DWARN("hostname is %s, arg is %s\n", host, addr);
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

int node_setstatus(const char *name, const char *from, const char *to)
{
        int ret;
        char path[MAX_PATH_LEN];

        ret = __node_checkname(name);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }


        if (strcmp(__node__.status, from)) {
                ret = EPERM;
                GOTO(err_ret, ret);
        }

        snprintf(path, MAX_NAME_LEN, "%s/config/name", __node__.home);

        ret = _set_value(path, name, strlen(name), O_TRUNC | O_SYNC);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        strcpy(__node__.name, name);

        snprintf(path, MAX_NAME_LEN, "%s/config/status", __node__.home);

        ret = _set_value(path, to, strlen(to), O_TRUNC | O_SYNC);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        strcpy(__node__.status, to);

        return 0;
err_ret:
        return ret;
}

static int __node_run_normal(etcd_lock_t *lock, const char *master, const int *idx)
{
        int ret;

        DINFO("========  run as normal ========\n");
        
        strcpy(__node__.status, NODE_STATUS_NORMAL);

        ret = netvip_destroy();
        if (unlikely(ret)) {
                DINFO("normal destroy vip ret( %d )\n", ret);
        }

        ret = node_normal_start(lock, master, idx);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

static int __node_run_normal__(etcd_lock_t *lock)
{
        int ret, idx;
        char master[MAX_NAME_LEN];
        uint32_t magic;

        ret = etcd_locker(lock, master, &magic, &idx);
        if (ret) {
                DWARN("get master fail %u %s\n", ret, strerror(ret));
                GOTO(err_ret, ret);
        }

        if (strcmp(master, __node__.name) == 0) {
                DWARN("lock fail, master %s node %s\n", master, __node__.name);
                ret = EAGAIN;
                GOTO(err_ret, ret);
        }

        DINFO("set master magic 0x%x\n", magic);
        ng.master_magic = magic;
        
        ret = __node_run_normal(lock, master, &idx);
        if (unlikely(ret)) {
                DWARN("normal fail %u %s\n", ret, strerror(ret));
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

static void *__node_start(void *_arg)
{
        int ret;
        etcd_lock_t lock;

        (void) _arg;

        strcpy(ng.name, __node__.name);

        ret = etcd_lock_init(&lock, ETCD_MISC, "master", gloconf.master_timeout, _random(), 1);
        if (ret)
                GOTO(err_ret, ret);

        while (1) {
                if (!etcd_is_proxy()) {
                        ret = etcd_lock(&lock);
                        if (ret == 0) {
                                ret = node_admin_start(&lock, __node__.name);
                                if (unlikely(ret)) {
                                        DWARN("master fail\n");
                                        sleep(1);
                                        continue;
                                }
                        } else if (ret != EEXIST) {
                                DWARN("new master fail %u %s\n", ret, strerror(ret));
                                sleep(1);
                                continue;
                        }

                }

                ret = __node_run_normal__(&lock);
                if (ret) {
                        sleep(1);
                        continue;
                }
        }

        return NULL;
err_ret:
        YASSERT(0);
        return NULL;
}

int node_start()
{
        int ret;

        YASSERT(strcmp(__node__.name, "none"));
        YASSERT(strcmp(__node__.status, NODE_STATUS_NONE));

        ret = sy_thread_create2(__node_start, NULL, "__node_start");
        if (unlikely(ret))
                GOTO(err_ret, ret);

        return 0;
err_ret:
        return ret;
}

int node_stat(nodestat_t *stat, nodedfree_t *node_dfree)
{
        int ret, i;
        uint64_t dfree = 0;
        static nodestat_t prev;
        static nodedfree_t prev_dfree;
        static time_t prev_load = 0;
        time_t now = 0;

        //memset(stat, 0x0, sizeof(*stat));
        if (__node__.inited == 0) {
                ret = EAGAIN;
                GOTO(err_ret, ret);
        }

        ret = sy_spin_lock(&__stat_lock__);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        now = gettime();
        if (prev_load != now || net_isnull(&prev.nid)) {
                prev.load = core_latency_get();
                ret = diskmd_node_dfree(&prev_dfree);
                if (unlikely(ret))
                        GOTO(err_lock, ret);

                //prev.used = prev.total - dfree;
                for (i = 0; i < prev_dfree.pool_count; i++) {
                        dfree += prev_dfree.pool_stat[i].total - prev_dfree.pool_stat[i].used;
                }

                prev.status = 0;
                if (node_ready())
                        prev.status |= __NODE_STAT_READY__;

                if (__node_deleting)
                        prev.status |= __NODE_STAT_DELETING__;

                if (node_ready())
                        prev.status |= __NODE_STAT_READY__;

#if 1
                if (dfree > cdsconf.disk_keep)
                        prev.status |= __NODE_STAT_WRITEABLE__;
#endif

                prev_load = now;

                prev.nid = *net_getnid();
        }

        *stat = prev;
        if (node_dfree)
                *node_dfree = prev_dfree;

        sy_spin_unlock(&__stat_lock__);
        return 0;
err_lock:
        sy_spin_unlock(&__stat_lock__);
err_ret:
        return ret;
}

int node_ready()
{
        //int count;
        nid_t admin;

        admin = *net_getadmin();

        if (net_isnull(&admin))
                return 0;

        if (net_islocal(&admin)) {
                return 1;
                //return node_writeable(&count);
        } else {
                return netable_connected(&admin);
        }
}

void node_set_deleting(int deleting)
{
        __node_deleting = deleting;
}

int node_get_deleting()
{
        //DINFO("deleting %u\n", __node_deleting);

        return __node_deleting;
}

int node_set_variable(const char *name, const char *key, const char *value)
{
	return  node_rpc_set_variable(name, key, value);
}

int node_fetch_variable(const char *name, const char *key, char *value)
{
        return node_rpc_fetch_variable(name, key, value);
}

int node_getinfo(nodeinfo_t *info, const char *name, char *buf)
{
        int ret, buflen;

        buflen = MAX_BUF_LEN;
        ret = node_rpc_none_getinfo(buf, &buflen, name, 1);
        if (unlikely(ret)) {
                if (ret == ENONET)
                        goto err_ret;
                else
                        GOTO(err_ret, ret);
        }

        nodeinfo_decode(buf, buflen, info);

        return 0;
err_ret:
        return ret;
}

int node_gethost(const char *admin, const nid_t *nid, char *host)
{
        return node_rpc_admin_gethost(admin, nid, host);
}

int node_getlist_open(const char *admin, const char *uuid)
{
        return node_rpc_admin_getlist_open(admin, uuid);
}

int node_getlist(const char *admin, char *buf, int *len, const char *uuid, uint64_t offset)
{
        return node_rpc_admin_getlist(admin, buf, len, uuid, offset);
}

int node_getlist_close(const char *admin, const char *uuid)
{
        return node_rpc_admin_getlist_close(admin, uuid);
}

int node_count(const char *admin, int *count)
{
        return node_rpc_admin_count(admin, count);
}

int node_init(const char *path)
{
        int ret;

        DINFO("node init\n");
        __node__.admin[0] = '\0';

        ret = sy_spin_init(&__stat_lock__);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ret = node_srv_init(path);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ret = node_rpc_init();
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ret = nodectl_init();
        if (unlikely(ret))
                GOTO(err_ret, ret);

        cluster_rpc_init();

        __node__.inited = 1;

        return 0;
err_ret:
        return ret;
}

int node_remove(const char *admin, const char *_name)
{
        return node_rpc_admin_dropnode(admin, _name);
}

int node_castoff(const char *name, int idx)
{
        return node_rpc_none_castoff(name, idx);
}

int node_pooldrop(const char *name, const char *pool)
{
        return node_rpc_none_pooldrop(name, pool);
}

int node_get_nid(const char *home)
{
        int ret;
        char path[MAX_PATH_LEN], tmp[MAX_NAME_LEN];
        struct stat stbuf;
        nid_t nid;

        ret = stat(home, &stbuf);
        if (ret < 0) {
                ret = errno;
                GOTO(err_ret, ret);
        }

        sprintf(path, "%s/data/node/config/nid", home);

        ret = _get_text(path, tmp, MAX_NAME_LEN);
        if (ret < 0) {
                ret = -ret;
                if (ret == ENOENT) {
                        memset(&nid, 0x0, sizeof(nid));
                } else
                        GOTO(err_ret, ret);
        }

        str2nid(&nid, tmp);

        YASSERT(nid.id >= 0);
        g_local_nid = nid;

        return 0;
err_ret:
        return ret;
}

int node_shutdown(void)
{
        netvip_destroy();
        rdma_running = 0;

        return 0;
}
